package cn.cl.lock1;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import static cn.cl.CurtorConfig.ADDRESS;

/**
 * 使用zookeeper实现分布式锁
 */
public class ZkLock {
    private static volatile ZkLock zkLock;

    protected static volatile CuratorFramework client=null;

    //当前节点
    private String currentLock;

    //等待节点
    private String waitPath;

    //父节点
    protected static final String PATH = "/lock";

    public ZkLock() {
        initClient();
    }

    /**
     * 初始化client
     */
    private void initClient() {
        if (client == null) {
            synchronized (ZkLock.class) {
                if (client == null) {
                    // 创建连接对象
                    client = CuratorFrameworkFactory
                            .builder()
                            // IP地址端口号
                            .connectString(ADDRESS)
                            // 会话超时时间
                            .sessionTimeoutMs(5000)
                            // 重连机制 重连3次，每次重连的间隔会越来越长
                            .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                            // 命名空间
                            // 构建连接对象
                            .build();
                    // 打开连接
                    client.start();
                }
            }
        }
    }


    /**
     * 获取锁
     */
    public void getLock() {
        if (tryLock()) {
            return;
        } else {
            waitLock();
            tryLock();
        }
    }

    private boolean tryLock() {
        //创建临时顺序节点
        try {
            currentLock=client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(PATH+"/data");
            System.out.println("currentLock:"+currentLock);
            //找到父节点下面的所有临时顺序节点
            List<String> childrens = client.getChildren().forPath(PATH);

            //判断当前的临时顺序节点是否是最小
            Collections.sort(childrens);
            for (String children : childrens) {
                System.out.println(children);
            }
            //获取到了锁
            if (currentLock.equals(PATH + "/" + childrens.get(0))) {
                return true;
            } else {
                int index = Collections.binarySearch(childrens, currentLock.substring(PATH.length() + 1));
                waitPath = childrens.get(index);
                return false;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }


        return false;
    }

    public static void main(String[] args) {
        ZkLock zkLock = new ZkLock();
        zkLock.tryLock();
    }

    /**
     * 等待解锁
     */
    private void waitLock() {
        try {
            CountDownLatch latch = new CountDownLatch(1);
            //监听上一节点
            PathChildrenCache nodeCache = new PathChildrenCache(client, PATH, true);
            nodeCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                        if (event.getData().getPath().equals(waitPath)) {
                            //解锁
                            latch.countDown();
                        }
                    }
                }
            });
            nodeCache.start();
            latch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 解锁
     */
    public void unLock() {
        try {
            client.delete().forPath(currentLock);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
